package kvraft

import (
	"6.5840/labgob"
	"6.5840/labrpc"
	"6.5840/raft"
	"bytes"
	"log"
	"sync"
	"sync/atomic"
	"time"
)

const (
	HandleOpTimeOut = time.Millisecond * 2000
)

const Debug = false

func DPrintf(format string, a ...interface{}) (n int, err error) {
	if Debug {
		log.Printf(format, a...)
	}
	return
}

type OpType string

const (
	OpGet    OpType = "Get"
	OpPut    OpType = "Put"
	OpAppend OpType = "Append"
)

type Op struct {
	// Your definitions here.
	OpCmd    OpType
	Key      string
	Value    string
	Seq      uint64
	ClientId int64
	// Field names must start with capital letters,
	// otherwise RPC will break.
}

type KVServer struct {
	mu      sync.Mutex
	me      int
	rf      *raft.Raft
	applyCh chan raft.ApplyMsg
	dead    int32 // set by Kill()

	waiCh      map[int]*chan Result // 等待commit信息的RPC handler通道
	historyMap map[int64]*Result    // 记录clerk的最高序列号和对应的结构result

	maxraftstate int               // the maximum allowed size of your persistent Raft state in bytes，including the log but not snapshot
	db           map[string]string // 存放key/value数据，进行持久化

	persister   *raft.Persister
	lastApplied int
}

type Result struct {
	LastSeq uint64
	Err     Err
	Value   string
	ResTerm int // 记录commit时被apply的Term，可能与Start时相比发生变化，要返回给客户端
}

// TODO: Handler RPC

func (kv *KVServer) HandleOp(opArgs *Op) (res Result) {
	kv.mu.Lock()
	if history, ok := kv.historyMap[opArgs.ClientId]; ok && history.LastSeq == opArgs.Seq {
		// 当前操作已经执行过，直接从操作历史中获取结果
		kv.mu.Unlock()
		DPrintf("Leader %v handle operation: Client %v Seq %v Request %s(%s, %s) from history.\n", kv.me, opArgs.ClientId, opArgs.Seq, opArgs.OpCmd, opArgs.Key, opArgs.Value)
		return *history
	}
	kv.mu.Unlock()
	DPrintf("Leader %v handle operation: Client %v Seq %v Request %s(%s, %s) calls Start\n", kv.me, opArgs.ClientId, opArgs.Seq, opArgs.OpCmd, opArgs.Key, opArgs.Value)

	startIndex, startTerm, isLeader := kv.rf.Start(*opArgs) //注意因为此处传参错误导致了msg.Command.(Op)中的应用错误
	if !isLeader {
		return Result{Err: ErrWrongLeader, Value: ""}
	}

	kv.mu.Lock()

	newCh := make(chan Result)
	kv.waiCh[startIndex] = &newCh
	DPrintf("Leader %v HandleOp: Client %v Seq %v Request: %s(%s, %s) new a chain %p\n", kv.me, opArgs.ClientId, opArgs.Seq, opArgs.OpCmd, opArgs.Key, opArgs.Value, &newCh)
	kv.mu.Unlock()

	defer func() {
		kv.mu.Lock()
		delete(kv.waiCh, startIndex)
		close(newCh)
		kv.mu.Unlock()
	}()

	select {
	case <-time.After(HandleOpTimeOut): // 将处理时间超时设置为2s
		res.Err = ErrHandleOpTimeOut
		DPrintf("server %v Client %v Seq %v Timeout\n", kv.me, opArgs.ClientId, opArgs.Seq)
		return
	case msg, success := <-newCh:
		if success && msg.ResTerm == startTerm {
			res = msg
			res.Err = OK
			DPrintf("server %v HandleOp: Client %v Seq %v handle operation %s(%s, %s) successfully, res=%s", kv.me, opArgs.ClientId, opArgs.Seq, opArgs.OpCmd, opArgs.Key, opArgs.Value, res.Value)
			return
		} else if !success {
			DPrintf("server %v HandleOp: Client %v Seq %v closed chain, args.OpCmd=%v, args.Key=%s", kv.me, opArgs.ClientId, opArgs.Seq, opArgs.OpCmd, opArgs.Key)
			res.Err = ErrChanClosed
			return
		} else {
			DPrintf("server %v HandleOp: Client %v Seq %v Leader Term changed, res.ResTerm=%v startTerm=%v", kv.me, opArgs.ClientId, res.ResTerm, res.ResTerm, startTerm)
			res.Err = ErrLeaderOutdated
			return
		}
	}
}

func (kv *KVServer) Get(args *GetArgs, reply *GetReply) {
	// Your code here.
	opArgs := &Op{
		OpCmd:    OpGet,
		Key:      args.Key,
		Seq:      args.Seq,
		ClientId: args.ClientId,
	}

	res := kv.HandleOp(opArgs)
	reply.Err = res.Err
	reply.Value = res.Value
}

func (kv *KVServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {
	// Your code here.
	opArgs := &Op{
		OpCmd:    OpPut,
		Key:      args.Key,
		Value:    args.Value,
		Seq:      args.Seq,
		ClientId: args.ClientId,
	}
	if args.Op == "Append" {
		opArgs.OpCmd = OpAppend
	}

	res := kv.HandleOp(opArgs)
	reply.Err = res.Err
}

func (kv *KVServer) execute() {
	for kv.killed() == false {
		msg := <-kv.applyCh
		if msg.CommandValid {
			DPrintf("")
			op := msg.Command.(Op)
			kv.mu.Lock()

			// 可能当前的log包含在快照中
			if msg.CommandIndex <= kv.lastApplied {
				kv.mu.Unlock() // 应该释放锁的地方未释放
				continue
			}
			kv.lastApplied = msg.CommandIndex

			res := Result{}
			needApply := false
			if history, ok := kv.historyMap[op.ClientId]; ok {
				if history.LastSeq == op.Seq {
					res = *history
				} else if history.LastSeq < op.Seq {
					needApply = true
				}
			} else {
				needApply = true
			}

			if needApply {
				res = kv.applyState(&op)
				res.ResTerm = msg.SnapshotTerm

				kv.historyMap[op.ClientId] = &res
			}

			ch, ok := kv.waiCh[msg.CommandIndex]
			if ok {
				kv.mu.Unlock()
				// 信息发送
				func() {
					defer func() {
						if recover() != nil {
							// 出现panic时说明通道关闭，进行日志打印
							DPrintf("")
						}
					}()
					res.ResTerm = msg.SnapshotTerm

					*ch <- res
				}()
				kv.mu.Lock()
			}

			if kv.maxraftstate != -1 && kv.persister.RaftStateSize() >= kv.maxraftstate/100*98 { // 如果到最大值快照，会发生log未截断的错误
				// 容量到98%时生成快照
				snapshot := kv.Snapshot()
				kv.rf.Snapshot(msg.CommandIndex, snapshot)
			}
			kv.mu.Unlock()
		} else {
			kv.mu.Lock()
			snapshot := msg.Snapshot
			if msg.SnapshotIndex >= kv.lastApplied { // 注意snapshotIndex和commitIndex（滥用自动补齐导致的错误）
				kv.readSnapshot(snapshot)
				kv.lastApplied = msg.SnapshotIndex
			}
			kv.mu.Unlock()
		}
	}
}

func (kv *KVServer) applyState(Operation *Op) Result {
	res := Result{}
	res.LastSeq = Operation.Seq
	switch Operation.OpCmd {
	case OpGet:
		val, ok := kv.db[Operation.Key]
		if ok {
			res.Err = OK
			res.Value = val
			DPrintf("")
			return res
		} else {
			res.Err = ErrNoKey
			res.Value = ""
			DPrintf("")
			return res
		}
	case OpPut:
		kv.db[Operation.Key] = Operation.Value
		res.Err = OK
		DPrintf("")
		return res
	case OpAppend:
		val, ok := kv.db[Operation.Key]
		res.Err = OK
		if ok {
			kv.db[Operation.Key] = val + Operation.Value
			DPrintf("")
			return res
		} else {
			kv.db[Operation.Key] = Operation.Value
			DPrintf("")
			return res
		}

	}

	return res
}

// TODO: SnapShot
func (kv *KVServer) Snapshot() []byte {
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)

	e.Encode(kv.db)
	e.Encode(kv.historyMap)

	serverSate := w.Bytes()
	return serverSate
}

func (kv *KVServer) readSnapshot(snapshot []byte) {
	if len(snapshot) == 0 {
		DPrintf("server %v read snapshot failed: snapshot is null", kv.me)
		return
	}

	r := bytes.NewBuffer(snapshot)
	d := labgob.NewDecoder(r)

	tmpDB := make(map[string]string)
	tmpHistoryMap := make(map[int64]*Result)

	if d.Decode(&tmpDB) != nil ||
		d.Decode(&tmpHistoryMap) != nil {
		DPrintf("server %v read snapshot failed", kv.me)
	} else {
		kv.db = tmpDB
		kv.historyMap = tmpHistoryMap
		DPrintf("server %v read snapshot succeed", kv.me)
	}
}

// the tester calls Kill() when a KVServer instance won't
// be needed again. for your convenience, we supply
// code to set rf.dead (without needing a lock),
// and a killed() method to test rf.dead in
// long-running loops. you can also add your own
// code to Kill(). you're not required to do anything
// about this, but it may be convenient (for example)
// to suppress debug output from a Kill()ed instance.
func (kv *KVServer) Kill() {
	atomic.StoreInt32(&kv.dead, 1)
	kv.rf.Kill()
	// Your code here, if desired.
}

func (kv *KVServer) killed() bool {
	z := atomic.LoadInt32(&kv.dead)
	return z == 1
}

// servers[] contains the ports of the set of
// servers that will cooperate via Raft to
// form the fault-tolerant key/value service.
// me is the index of the current server in servers[].
// the k/v server should store snapshots through the underlying Raft
// implementation, which should call persister.SaveStateAndSnapshot() to
// atomically save the Raft state along with the snapshot.
// the k/v server should snapshot when Raft's saved state exceeds maxraftstate bytes,
// in order to allow Raft to garbage-collect its log. if maxraftstate is -1,
// you don't need to snapshot.
// StartKVServer() must return quickly, so it should start goroutines
// for any long-running work.
func StartKVServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister, maxraftstate int) *KVServer {
	// call labgob.Register on structures you want
	// Go's RPC library to marshall/unmarshall.
	labgob.Register(Op{})

	kv := new(KVServer)
	kv.me = me
	kv.maxraftstate = maxraftstate

	// You may need initialization code here.
	kv.persister = persister

	kv.applyCh = make(chan raft.ApplyMsg)
	kv.rf = raft.Make(servers, me, persister, kv.applyCh)

	// You may need initialization code here.
	kv.historyMap = make(map[int64]*Result)
	kv.db = make(map[string]string)
	kv.waiCh = make(map[int]*chan Result)

	// 启动时通过快照启动
	kv.mu.Lock()
	kv.readSnapshot(persister.ReadSnapshot())
	kv.mu.Unlock()

	go kv.execute()

	return kv
}
